package com.project.rfidtodb.service.Impl;


import cn.hutool.core.collection.CollectionUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.project.rfidtodb.constant.Constants;
import com.project.rfidtodb.entity.TbCommandA0;
import com.project.rfidtodb.entity.TbCommandA0His;
import com.project.rfidtodb.entity.TbCommandA1;
import com.project.rfidtodb.mapper.TbCommandA0HisMapper;
import com.project.rfidtodb.mapper.TbCommandA0Mapper;
import com.project.rfidtodb.service.ProcessService;
import com.project.rfidtodb.service.TbCommandA0Service;
import com.project.rfidtodb.util.SpringContextUtil;
import com.project.util.JSONUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;

@Slf4j
@Service
public class TbCommandA0ServiceImpl extends ServiceImpl<TbCommandA0Mapper, TbCommandA0> implements TbCommandA0Service {

    @Resource
    private Executor taskExecutor;

    @Resource
    @Lazy
    private TbCommandA0Service selfService;

    @Resource
    private TbCommandA0HisMapper tbCommandA0HisMapper;
    /**
     * 获取实时表所偶记录
     * @return
     */
    @DS("master")
    public List<TbCommandA0> selectByPage(IPage<TbCommandA0> page, LambdaQueryWrapper<TbCommandA0> queryWrapper){
        log.info("selectByPage invoke......");
        IPage<TbCommandA0> pageResult = super.page(page, queryWrapper);
        return pageResult.getRecords();
    }


    /**
     * 获取实时表所偶记录
     * @return
     */
    @DS("master")
    public List<TbCommandA0> selectAll( ){
        log.info("selectAll invoke......");
        return super.list();
    }
    /**
     * 插入Sqlserver数据库
     * @param list
     */
    @DS("sqlserver")
    public void insertSqlServer(List<TbCommandA0> list){
        log.info("insertSqlServer invoke......");
        String currentDataSource = DynamicDataSourceContextHolder.peek();
        log.info("Current data source: {}", currentDataSource);
        if(CollectionUtil.isNotEmpty(list)) {
//            list.forEach(item->taskExecutor.execute(()->baseMapper.insertWithoutKeyReturn(item)));
            baseMapper.insertBatchWithoutKeyReturn(list);

        }
    }
    /**
     * 插入kingbase数据库
     * @param list
     */
    @DS("kingbasedb")
    public void insertKingbase(List<TbCommandA0> list){
        log.info("insertKingbase invoke......");
        String currentDataSource = DynamicDataSourceContextHolder.peek();
        log.info("Current data source: {}", currentDataSource);
        if(CollectionUtil.isNotEmpty(list)) {
//            list.forEach(item->taskExecutor.execute(()->baseMapper.insertWithoutKeyReturn(item)));
            baseMapper.insertBatchWithoutKeyReturn(list);

        }
    }

    /**
     * 清空实时表记录
     */
    @DS("master")
    public void delete(List<TbCommandA0> list){
        log.info("delete invoke......");
        String currentDataSource = DynamicDataSourceContextHolder.peek();
        log.info("Current data source: {}", currentDataSource);
        if(CollectionUtil.isNotEmpty( list)) {
            super.removeBatchByIds(list.stream().map(TbCommandA0::getSequenceNumber).collect(Collectors.toList()));
        }
    }
    /**
     * 加入历史表记录
     * @param list
     */
    @DS("master")
    public void insertHis(List<TbCommandA0> list){
        log.info("insertHis invoke......");
        //删除30天前的数据
        tbCommandA0HisMapper.delete(Wrappers.<TbCommandA0His>lambdaQuery().lt(TbCommandA0His::getCreateTime, LocalDateTime.now().plusMonths(-1)));
       //保存历史归档
        if(CollectionUtil.isNotEmpty(list)) {
            List<TbCommandA0His> hisList = list.stream().map(item -> {
                TbCommandA0His his = new TbCommandA0His();
                BeanUtils.copyProperties(item, his);
                return his;
            }).collect(Collectors.toList());
            tbCommandA0HisMapper.insertBatchWithoutKeyReturn(hisList);
        }
    }
    /**
     * 操作过程
     */
    @Async("taskExecutor")
    public void process() throws Exception{
        log.info("TbCommandA0 process invoke......");
        long count = 0;
        long current = 1;
//        List<TbCommandA0> list = null;
        IPage<TbCommandA0> page = new Page<>(current, Constants.size);
        while (true) {
            try {
                log.info("TbCommandA0 current:{}",current);
                //获取实时表数据
                List<TbCommandA0> list = selfService.selectByPage(page, new LambdaQueryWrapper<TbCommandA0>().orderByDesc(TbCommandA0::getSequenceNumber));
                if (CollectionUtil.isNotEmpty(list)) {
//                //插入Sqlserver数据库
//                    selfService.insertSqlServer(list);
//                //插入kingbase数据库
//                    selfService.insertKingbase(list);

                    // 使用并行流同时执行多个数据库操作
                    CompletableFuture<Void> sqlServerFuture = CompletableFuture.runAsync(() -> {
                        selfService.insertSqlServer(list);
                    });

                    CompletableFuture<Void> kingbaseFuture = CompletableFuture.runAsync(() -> {
                        selfService.insertKingbase(list);
                    });
                    //加入历史表备份
                    CompletableFuture<Void> hisFuture = CompletableFuture.runAsync(() -> {
                        selfService.insertHis(list);
                    });

                    // 等待插入操作完成后再执行删除和历史记录操作
                    CompletableFuture.allOf(sqlServerFuture, kingbaseFuture,hisFuture).join();


                    //清空实时表记录
                    selfService.delete(list);

                    count = count + list.size();
                    log.info("TbCommandA0 data current count:{}", count);
//                    current++;
//                    page.setCurrent(current);
//                list = null;
                }
//                else {
//                    log.info("TbCommandA0 data total count:{}", count);
//                    break;
//                }

            }catch (Exception e){
                log.error("TbCommandA0 process error:{}",e);
            }
        }



    }
}
